package com.kamistoat.rocketmqlearning.generalMsg;

import com.kamistoat.rocketmqlearning.common.R;
import com.kamistoat.rocketmqlearning.msgProducer.MsgProducerFactory;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import java.nio.charset.StandardCharsets;

@RequestMapping("/generalMsg")
@Controller
public class GeneralMsgController {

    @Autowired
    @Qualifier("msgProducer")
    DefaultMQProducer msgProducer;

    @Autowired
    MsgProducerFactory msgProducerFactory;

    /**
     * 同步发送消息
     */
    @RequestMapping("/sync")
    @ResponseBody
    public R syncProducer() {
        try {
            // 发送消息前开启生产者.官方建议一个线程内所有消息发送完后再关闭生产者，不要频繁开关，这会导致创建大量低效的短连接
            // DefaultMQProducer msgProducer = msgProducerFactory.newMsgProducer();
            // msgProducer.start();

            // 而当使用单例Bean来管理时，务必停止手动start和shutdown，而是由Bean生命周期接手

            // 使用生产者发送10条消息
            for (int i = 0; i < 20; i++) {
                /**
                 * 使用RocketMQ库类创建消息
                 */
                Message mQMsg = new Message();
                /**
                 * 绑定消息的Topic。如果没有提前创建则Broker会自行创建
                 */
                mQMsg.setTopic("someTopic");
                /**
                 * 绑定消息的Tags
                 */
                mQMsg.setTags("someTag");
                /**
                 * 绑定消息的Key。Key可以用来作为幂等消费的唯一性标识
                 */
                mQMsg.setKeys("Key-" + i);
                /**
                 * RocketMQ必须以byte[]发送消息。而不能向RabbitMQ一样直接以Object发送.
                 * 通常的做法是将Object转成JSONString，然后再将String制作成byte[]进行发送.
                 */
                byte[] body = ("Hi" + i).getBytes(StandardCharsets.UTF_8);
                mQMsg.setBody(body);

                /**
                 * Broker返回SendResult类显示发送是否成功
                 */
                SendResult sendResult = this.msgProducer.send(mQMsg);
                System.out.println(sendResult);
            }

            return R.ok();
        } catch (MQClientException | MQBrokerException | RemotingException | InterruptedException e) {
            e.printStackTrace();
            return R.error(e.getMessage());
        } finally {
            // 关闭生产者
            // msgProducer.shutdown();
        }
    }

    /**
     * 异步发送消息
     */
    @RequestMapping("/async")
    @ResponseBody
    public R asyncProducer() {
        try {
            // DefaultMQProducer msgProducer = msgProducerFactory.newMsgProducer();
            // msgProducer.start();

            for (int i = 0; i < 20; i++) {
                Message mQMsg = new Message();
                mQMsg.setTopic("someTopic");
                mQMsg.setTags("someTag");
                mQMsg.setKeys("Key-" + i);
                byte[] body = ("Hi" + i).getBytes(StandardCharsets.UTF_8);
                mQMsg.setBody(body);

                /**
                 * 异步发送消息需要手写回调函数.
                 * 仍然是使用send方法，但是使用可添加SendCallback回调的重载.
                 * 生产者不会等待Broker的返回结果，但是当Broker返回时会触发对应的回调函数
                 */
                msgProducer.send(mQMsg, new SendCallback() {
                    /**
                     * 当Broker返回SendResult时就触发。只要有SendResult，不管sendStatus是啥都会触发
                     */
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        System.out.println(sendResult);
                    }

                    /**
                     * 否则触发异常回调
                     */
                    @Override
                    public void onException(Throwable e) {
                        e.printStackTrace();
                    }
                });
            }
            /**
             * 异步发送消息时，在关闭producer之前，应该等待 2*(生产者到Broker网络延时) 的时间
             * 目的是为了等待所有消息都发送完毕且异步回调也全部执行完。
             */
            Thread.sleep(10);
            return R.ok();
        } catch (MQClientException | RemotingException | InterruptedException e) {
            e.printStackTrace();
            return R.error(e.getMessage());
        } finally {
            // 关闭生产者
            // msgProducer.shutdown();
        }
    }
}
